package com.framework.flink;

import org.apache.flink.api.common.JobExecutionResult;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSink;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;

/**
 * 描述：
 * 数据写入
 * 基于DataStream
 * @author xianggj
 * @Date 2021/10/19 14:31
 **/
public class FlinkTestMysqlSink {
    public static void main(String[] args) throws Exception {
        // 调用程序
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 添加自定义数据源

        DataStreamSink<Tuple2<String, String>> tuple2DataStreamSink = env.addSource(new FlinkTestMysqlSource.MySourceMysql())
                .addSink(new MySinkMysql());
        // 提交执行任务
        JobExecutionResult result = env.execute("MySinkMysql");
        System.out.println(result.getJobID());

    }

    public static class MySinkMysql extends RichSinkFunction<Tuple2<String, String>> {
        private static final String UPSERT_CASE = "INSERT INTO test1 (id, name) VALUES (?, ?) ";

        private static final String driver = "com.mysql.jdbc.Driver";
        private static final String url = "jdbc:mysql://192.168.1.37:3306/test";
        private static final String username = "root";
        private static final String password = "123456";
        private PreparedStatement statement;
        private Connection connection = null;


        @Override
        public void invoke(Tuple2<String, String> value, Context context) throws Exception {
            statement.setString(1, value.f0);
            statement.setString(2, value.f1);
//            statement.setString(3, aCase.getTraceHash());
            statement.addBatch();
            statement.executeBatch();
        }

        @Override
        public void open(Configuration parameters) throws Exception {
            Class.forName(driver);
            connection =
                    DriverManager.getConnection(url, username, password);

            statement = connection.prepareStatement(UPSERT_CASE);
        }

        @Override
        public void close() throws Exception {
            super.close();
            if(connection != null){
                connection.close();
            }
            if (statement != null){
                statement.close();
            }
        }
    }
}
